use std::{
	net::IpAddr,
	sync::atomic::{AtomicU32, Ordering},
	time::Duration,
};

use base_data::channel::{
	event::ChannelEvent, ChannelCallDirection, CommandMsg, EndReason, OriginateChannel, OriginateMsg,
};
use bytes::{Bytes, BytesMut};
use tokio::{
	sync::mpsc::{self, error::TrySendError},
	time,
};
use tools::{error::Error, utils::network::combination_address_replace, ResultEx, ResultNull};
use tracing::{error, info, trace, warn};

use crate::{
	rtp::{
		event::RtpEvent,
		manager::{generate_rtp_listen, start_rtp_run_control},
	},
	sdp::{local_sdp, Sdp},
	sip2::{
		config::sip_config,
		head::{
			SipHeadField, ALLOW, ALLOW_EVENTS, CONTACT, CONTENT_LENGTH, CONTENT_TYPE, FROM, MAX_FORWARDS,
			P_EARLY_MEDIA, SUPPORTED, TO,
		},
		manage::to_distribute,
		transport::TransportProtocol,
		utils::{
			cseq_num, generate_full_sip_url, generate_tag_value, get_bytes_str, position_to_tag, WriteSip,
		},
		SipMethod, SipResponse,
	},
	wav::control::load_wav_bufs,
	Codec,
};

use super::{
	dialog_iden::DialogIdent,
	sipdata::SipData,
	transaction::{SipTimeOver, SipTransaction, SipTransactionState, WaitSipMessage},
	DialogEvent, DialogMessage, LENGTH_ZERO,
};

#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone, Copy)]
pub(crate) enum DialogState {
	Init,
	Ringing,
	PreAnswer,
	Answered,
	///此值代表在当前的一个事务当中.收到了需要挂断.但事务已经进入到确认阶段,必须等确认完成再进行
	NeedKill,
	Hangup,
}
///用来在分发处理者这边保存对话信息.以便于查找和发送消息
#[derive(Debug)]
pub struct Dialog {
	pub(crate) id: u32,
	dialog_ident: DialogIdent,
	pub(crate) from_business: mpsc::Receiver<CommandMsg>,
	pub(crate) from_sip_dist: mpsc::Receiver<DialogMessage>,

	to_business: mpsc::Sender<ChannelEvent>,
	///给dialog管理器的发送器.
	to_manager: mpsc::Sender<DialogEvent>,
	to_rtp: mpsc::Sender<RtpEvent>,

	///通过传输层向远端发送消息的对象.
	pub(crate) transport: TransportProtocol,
	///当前dialog对应的是方向是入站还是出站..借用通道的方向
	pub(crate) call_direction: ChannelCallDirection,

	pub(crate) sip_data: SipData,
	pub(crate) caller_num: Bytes,
	pub(crate) dest_num: Bytes,
	pub(crate) local_addr: Bytes,
	// pub(crate) peer_addr: Bytes,
	sdp: Sdp,
	sip_to_tag: Bytes,
	sip_to_name: Bytes,
	pub(crate) state: DialogState,
	///是否可以发送媒体数据
	// is_can_send_media: bool,
	///保存最后一次发送的数据包.用来重试发送
	pub(crate) last_send_bytes: Option<Bytes>,
	pub(crate) wait_resp_method: WaitSipMessage,

	pub(crate) retry_time: Duration,
	pub(crate) start_time: Duration,
	///呼叫超时时间,如果需要设定业务逻辑呼叫时间,同样设置此值.
	pub(crate) over_time: SipTimeOver,

	tran_state: SipTransactionState,
	pub(crate) tran_handler: SipTransaction,
	///是否成功建立会话
	// pub(crate) success_create_session: bool,
	///此值为true,代表在当前的一个事务当中.收到了需要挂断.但事务已经进入到确认阶段,必须等确认完成再进行.
	// pub(crate) need_kill: bool,
	///当前已经使用的cseq的值
	pub(crate) cseq: u32,
}

impl Drop for Dialog {
	fn drop(&mut self) {
		trace!("dialog 结束");
		//发送给Dialog管理者,当前dialog退出消息
		//这里不需要再向rtp发送结束信号,因为rtp在接收消息时会自动发现发送者关闭了
		self.send_to_manager(DialogEvent::Close(self.id));
		self.send_to_rtp(RtpEvent::DialogClose(self.id));
	}
}
impl Dialog {
	///从传输层创建一个对话.
	/// 这个应该只有呼入的第一次才创建
	pub(crate) fn from_sip(
		id: u32,
		sip_data: SipData,
		transport: TransportProtocol,
		to_business: mpsc::Sender<ChannelEvent>,
		from_business: mpsc::Receiver<CommandMsg>,
		from_sip_dist: mpsc::Receiver<DialogMessage>,
		sender_transaction: mpsc::Sender<DialogEvent>,
		rtp_to_dialog: mpsc::Sender<DialogMessage>,
	) -> ResultEx<Self> {
		let (caller_num, _) = sip_data.caller_num();
		let (dest_num, local_addr) = sip_data.dest_num();
		let sdp = match Sdp::from_bytes(&mut sip_data.body().clone()) {
			Ok(s) => s,
			Err(Error::MediaError(e)) => {
				error!("解码sdp失败:{}", e);
				let buf = sip_data.generate_unsupported_media_type();
				return Err(Error::SipErrorRecover(buf));
			}
			Err(Error::NotFound(_)) => {
				trace!("没有找到sdp消息.给自己默认值");
				local_sdp()
			}
			Err(e) => return Err(e),
		};

		let ip = transport.local_addr().1.ip();
		let (rtp_sock, rtcp_sock) = generate_rtp_listen(ip)?;
		let mut merge_sdp = sdp.merge(&local_sdp())?;
		merge_sdp.set_addr(
			ip,
			rtp_sock.local_addr()?.port(),
			rtcp_sock.local_addr()?.port(),
		);
		let (to_name, to_tag) = position_to_tag(sip_data.to.bytes());

		Ok(Self {
			dialog_ident: DialogIdent::new(id, sip_data.call_id.bytes().clone()),
			to_business,
			from_business,
			from_sip_dist,
			to_manager: sender_transaction,
			to_rtp: start_rtp_run_control(
				id,
				rtp_to_dialog,
				combination_address_replace(
					sdp.c.address,
					&transport.peer_addr().ip(),
					sdp.first_rtp_port()
				),
				rtp_sock,
				rtcp_sock,
				&merge_sdp.m,
			)?,
			transport,
			id,
			call_direction: ChannelCallDirection::InBound,
			cseq: cseq_num(sip_data.cseq.bytes()),
			sip_data,
			caller_num,
			dest_num,
			local_addr,
			sdp: merge_sdp,
			sip_to_tag: to_tag,
			sip_to_name: to_name,
			last_send_bytes: Default::default(),
			retry_time: Duration::ZERO,
			start_time: Default::default(),
			over_time: SipTimeOver::Zero,
			wait_resp_method: WaitSipMessage::None,
			tran_state: SipTransactionState::Terminated,
			tran_handler: SipTransaction::InviteServer,
			state: DialogState::Init,
		})
	}

	///判断当前连接是否安全的协议
	#[inline]
	pub(crate) fn is_unsafe_protocol(&self) -> bool {
		match self.transport {
			TransportProtocol::Udp(..) => true,
			_ => false,
		}
	}
	pub(crate) fn ident(&self) -> &DialogIdent {
		&self.dialog_ident
	}

	///启动一个dialog的运行.
	pub(crate) async fn start_run(mut self) {
		let mut time_over;
		let mut tran = self.tran_handler;
		tran.start_tran(&mut self);

		let sleep = time::sleep(Duration::ZERO);
		tokio::pin!(sleep);
		loop {
			//事务结束的处理.结束处理后再进是否换事务
			if self.tran_state == SipTransactionState::Terminated {
				match tran.end_tran(&mut self) {
					//事务退出,并且dialog同时退出
					Ok(true) => break,
					//当前事务退出,由事务自己指明下一个操作方式
					Ok(false) => {}
					Err(e) => error!("处理超时失败:{}", e),
				}
			}

			//如果有需要替换的对象,则进行替换
			if tran != self.tran_handler {
				tran = self.tran_handler;
				tran.start_tran(&mut self);
			}

			//得到本次循环的超时时间...如果得到zero,不再进入循环,直接处理超时
			time_over = tran.timeout_period(&mut self);
			if time_over == SipTimeOver::Zero || time_over.time() == Duration::ZERO {
				if let Err(e) = tran.handle_time_over(&mut self, time_over) {
					error!("处理超时失败:{}", e);
				}

				continue;
			}
			trace!("等待超时时间:{:?}", time_over);
			sleep.set(time::sleep(time_over.time()));

			//只使用一个超时器.超时原因,在打开的时候进行判断
			tokio::select! {
				biased;
				Some(sd) = self.from_sip_dist.recv() => {
					match sd{
						DialogMessage::Dtmf(dtmf) => {
							self.send_to_business(ChannelEvent::DTMF(self.id,dtmf));
						}
						DialogMessage::SipData(s) => {
							match tran.handle_tran_msg(&mut self, s) {
								Ok(_) => {}
								Err(Error::NotFound(s)) => { //出现没有匹配对应的媒体
									error!(s);
									let buf = self.generate_unsupported_media_type();
									self.send_to_remote( buf);
									self.send_to_business(ChannelEvent::ChannelHangup(self.id,EndReason::SissionError));
									break;
								}
								Err(e) => {
									error!("接收处理sip数据失败:{}",e);
								}
							}
						}
						DialogMessage::StopPlay => self.send_to_business(ChannelEvent::PlaybackStop(self.id,)),
					}
					//所有收到的sip消息都先交给事务..再根据情况处理

				}
				Some(cm) = self.from_business.recv() => {
					//根据情况处理,有部分不发送至sip的事务处理
					match cm {
						CommandMsg::Originate(m) => {
							info!("开始一个呼出动作,创建dialog");
							let transport = match TransportProtocol::from_call_way(&m).await {
								Ok(t) => t,
								Err(Error::UserNotRegister) => {
									info!("用户未注册");
									self.send_to_business(ChannelEvent::OriginateError(EndReason::UserNotRegister));
									continue;
								}
								Err(e) => {
									info!("拿sip地址失败:{}",e);
									self.send_to_business(ChannelEvent::OriginateError(EndReason::ChannelUnusable));
									continue;
								}
							};

							let (s_d, r_d) = mpsc::channel(2048);
							let (s, r) = mpsc::channel(2048);
							let (dist_t_dialog, dist_r_dialog) = mpsc::channel(2048);
							let dialog_id = dialog_id();

							let channel = OriginateChannel {
								id: dialog_id,
								to_switch: s_d,
								from_switch: r,
							};

							let ip = transport.local_addr().1.ip();
							let dialog = match Self::from_originate(
								dialog_id,
								&m,
								transport,
								r_d,
								dist_r_dialog,
								s,
								self.to_rtp.clone(),
								ip,
								&self.sdp) {
									Ok(d) => d,
									Err(e) => {
										error!("创建dialog失败:{}",e);
										self.send_to_business(ChannelEvent::OriginateError(EndReason::SissionError));
										continue;
									}
							};
							//先给逻辑发送一个开始呼叫事件
							self.send_to_business(ChannelEvent::ChannelOriginate(channel));
							//将两端发送相应信息
							self.send_to_rtp(RtpEvent::SetBLag(dialog.id, dist_t_dialog.clone(), ));
							self.send_to_manager(DialogEvent::NewDialog(dialog, dist_t_dialog));
						}
						CommandMsg::PreAnswer => self.pre_answer(),
						CommandMsg::PlayBell(file_path) => {
							self.pre_answer();

							match load_wav_bufs(&file_path).await {
								Ok(wb) => self.send_to_rtp(RtpEvent::PlayFile(self.id,wb)),
								Err(e) => error!("加载文件失败,file:{},,{}",&file_path,e)
							}
						}
						CommandMsg::PlayBellList(file_paths) => {
							self.pre_answer();

							let mut wbs = Vec::with_capacity(file_paths.len());
							for file_path in file_paths.iter() {
								match load_wav_bufs(file_path).await {
									Ok(wb) => wbs.push(wb),
									Err(e) => error!("加载文件失败,file:{},,{}",&file_path,e)
								}
							}

							self.send_to_rtp(RtpEvent::PlayListFiles(self.id,wbs));
						}
						//以下这些是送给rtp协程处理的
						CommandMsg::StopPlayBell =>	self.send_to_rtp(RtpEvent::StopPlay(self.id)),
						CommandMsg::BridgeChannel(_other_id) => self.send_to_rtp(RtpEvent::Bind),
						CommandMsg::StartRecordSound(file_path) => self.send_to_rtp(RtpEvent::StartRecordFile(file_path)),
						CommandMsg::StopRecordSound => self.send_to_rtp(RtpEvent::StopRecord),

						//与事务相关事务处理
						CommandMsg::Ringing |
						CommandMsg::AnswerCall(_) |
						CommandMsg::KillChannel(_) |
						CommandMsg::HangupChannel(_, _) =>
							if let Err(e) = tran.handle_command_msg(&mut self, cm) {
								error!("接收处理命令失败:{}",e);
							}

						//目前不处理
						CommandMsg::SetVar|
						CommandMsg::SendDtmf(_, _) => {},
					}
				}
				//处理各种超时状态
				_ = &mut sleep => {
					if let Err(e) = tran.handle_time_over(&mut self, time_over) {
						error!("处理超时失败:{}",e);
					}
				}
			}
		}
	}

	///向sip消息的网络端发送一个数据.此数据会记录下来.当需要重发时使用
	#[inline]
	pub(crate) fn send_to_remote(&mut self, bytes: Bytes) {
		self.transport.send(bytes.clone());
		self.last_send_bytes = Some(bytes);
	}

	pub(crate) fn retry_send(&self) -> ResultNull {
		if let Some(buf) = self.last_send_bytes.as_ref() {
			self.transport.send(buf.clone());
			Ok(())
		} else {
			Err(Error::LogicalError("尝试重发时,没有发送过数据"))
		}
	}

	pub(crate) fn from_originate(
		id: u32,
		m: &OriginateMsg,
		transport: TransportProtocol,
		from_business: mpsc::Receiver<CommandMsg>,
		from_sip_dist: mpsc::Receiver<DialogMessage>,
		to_business: mpsc::Sender<ChannelEvent>,
		to_rtp: mpsc::Sender<RtpEvent>,
		ip: IpAddr,
		sdp: &Sdp,
	) -> ResultEx<Self> {
		let sip_data = Self::generate_invite(m, sdp, &transport);
		let (caller_num, _) = sip_data.caller_num();
		let (dest_num, local_addr) = sip_data.dest_num();

		let (to_name, to_tag) = position_to_tag(sip_data.to.bytes());
		//将呼叫过来的sdp,换ip和监听端口.这样可以直接把呼入的sdp发送给对方.能保证id是一致的
		let mut sdp = sdp.clone();
		sdp.set_ip(ip);

		Ok(Self {
			id,
			last_send_bytes: Some(sip_data.data.clone()),
			dialog_ident: DialogIdent::new(id, sip_data.call_id.bytes().clone()),
			sip_data,
			to_business,
			from_business,
			from_sip_dist,
			to_manager: to_distribute().clone(),
			transport,
			call_direction: ChannelCallDirection::OutBound,
			caller_num,
			dest_num,
			local_addr,
			sdp,
			to_rtp,
			sip_to_tag: to_tag,
			sip_to_name: to_name,
			retry_time: Duration::ZERO,
			start_time: Default::default(),
			over_time: SipTimeOver::Zero,
			wait_resp_method: WaitSipMessage::None,
			tran_state: SipTransactionState::Terminated,
			tran_handler: SipTransaction::InviteClient,
			cseq: 1,
			state: DialogState::Init,
		})
	}

	///生成一个invite消息
	pub(crate) fn generate_invite(
		msg: &OriginateMsg,
		sdp: &Sdp,
		transport: &TransportProtocol,
	) -> SipData {
		let mut buf = BytesMut::with_capacity(2048);
		let mut sd = SipData::default();
		sd.peer_adder = transport.peer_addr();
		sdp.write_to(&mut buf);
		sd.body = buf.split().freeze();

		let (tp_name, ip) = transport.local_addr();
		sd.title_name = format!("sip:{}@{}", &msg.called_num, &ip.ip().to_string()).into();
		sd.cseq = SipHeadField::generate_cseq(SipMethod::Invite);
		sd.allow = SipHeadField::Allow(Bytes::from_static(sip_config().allow.as_bytes()));
		sd.max_forwards = SipHeadField::MaxForwards(Bytes::from_static(b"70"));
		sd.allow_events =
			SipHeadField::AllowEvents(Bytes::from_static(sip_config().allow_events.as_bytes()));
		//现在是固定一个呼叫类型
		sd.content_type = SipHeadField::ContentType(Bytes::from_static(b"application/sdp"));
		sd.supported = SipHeadField::Supported(Bytes::from_static(sip_config().supported.as_bytes()));
		sd.session_expires =
			SipHeadField::SessionExpires(Bytes::from_static(sip_config().session_expires.as_bytes()));
		sd.min_se = SipHeadField::MinSE(Bytes::from_static(sip_config().min_se.as_bytes()));
		sd.via = SipHeadField::generate_via(tp_name, &ip);
		sd.call_id = SipHeadField::generate_callid();
		sd.from = SipHeadField::generate_from(&msg.caller_name, &msg.caller_num, &ip);
		sd.to = SipHeadField::generate_to("", &msg.called_num, &sd.peer_adder);
		sd.p_asserted_identity =
			SipHeadField::generate_p_asserted_identity(&msg.caller_name, &msg.caller_num, &ip);
		sd.contact = SipHeadField::genterate_contact(&msg.caller_num, &ip);
		sd.content_length = SipHeadField::genterate_length(sd.body.len());

		//目前没有route和recode-route的字段值.
		buf.put_sip_request(SipMethod::Invite, &sd.title_name);
		buf.put_unbroken_head(&sd.via);
		buf.put_unbroken_head(&sd.route);
		buf.put_unbroken_head(&sd.from);
		buf.put_unbroken_head(&sd.to);
		buf.put_unbroken_head(&sd.call_id);
		buf.put_unbroken_head(&sd.cseq);
		buf.put_unbroken_head(&sd.allow);
		buf.put_unbroken_head(&sd.allow_events);
		buf.put_unbroken_head(&sd.contact);
		buf.put_unbroken_head(&sd.max_forwards);
		buf.put_unbroken_head(&sd.supported);
		buf.put_unbroken_head(&sd.session_expires);
		buf.put_unbroken_head(&sd.min_se);
		buf.put_unbroken_head(&sd.p_asserted_identity);
		buf.put_unbroken_head(&sd.content_length);
		buf.put_unbroken_head(&sd.content_type);
		buf.put_heads_end();
		buf.extend_from_slice(&sd.body);

		sd.data = buf.freeze();
		sd
	}

	///生成一个成功的Ack消息.
	///
	/// 这个消息在收到对方的200以后发送.目前不会变化发送方向.
	pub(crate) fn generate_inv_ack(&self) -> Bytes {
		//生成一个inv_ack消息
		let sip_data = &self.sip_data;
		let mut buf = BytesMut::with_capacity(1024);

		//第一行数据
		buf.put_sip_request(SipMethod::Ack, &sip_data.title_name);
		buf.put_unbroken_head(&sip_data.via);
		buf.put_unbroken_head(&sip_data.route);
		buf.put_unbroken_head(&sip_data.from);
		buf.put_unbroken_head(&sip_data.to);
		buf.put_unbroken_head(&sip_data.call_id);
		buf.put_unbroken_head(&sip_data.contact);
		buf.put_unbroken_head(&sip_data.max_forwards);
		buf.put_cseq_head(self.cseq, SipMethod::Ack.as_bytes());
		buf.put_head_field(CONTENT_LENGTH, LENGTH_ZERO);
		buf.put_heads_end();

		buf.freeze()
	}

	///准备向远端发送一个取消的消息...
	///
	/// 这个消息应该是在invite 过程当中处理的.所以取第一个的值进行匹配
	pub(crate) fn generate_cancel(&self) -> Bytes {
		// 生成一个Cancel的消息.这个要根据上一个消息内容生成
		let sip_data = &self.sip_data;

		let mut buf = BytesMut::with_capacity(1024);

		//第一行数据
		buf.put_sip_request(SipMethod::Cancel, &sip_data.title_name);
		buf.put_unbroken_head(&sip_data.via);
		buf.put_unbroken_head(&sip_data.route);
		buf.put_unbroken_head(&sip_data.from);
		buf.put_unbroken_head(&sip_data.to);
		buf.put_unbroken_head(&sip_data.call_id);
		buf.put_head_field(MAX_FORWARDS, b"70");
		buf.put_cseq_head(self.cseq, SipMethod::Cancel.as_bytes());

		//单独写入长度
		buf.put_head_field(CONTENT_LENGTH, LENGTH_ZERO);
		buf.put_heads_end();

		buf.freeze()
	}

	/// 生成一个Bye的消息...两边都可能生成
	pub(crate) fn generate_new_bye(&mut self) {
		let sip_data = &self.sip_data;

		let mut buf = BytesMut::with_capacity(1024);

		//第一行数据
		let (tp_name, ip) = self.transport.local_addr();
		let (to, from, via) = match self.call_direction {
			ChannelCallDirection::InBound | ChannelCallDirection::UnknownValue => (
				&sip_data.from,
				&sip_data.to,
				SipHeadField::generate_via(tp_name, &ip),
			),
			ChannelCallDirection::OutBound => (&sip_data.to, &sip_data.from, sip_data.via.clone()),
		};

		buf.put_sip_request(SipMethod::Bye, format!("sip:mod@{}", ip).as_bytes());
		buf.put_unbroken_head(&via);
		buf.put_unbroken_head(&sip_data.route);
		buf.put_head_field(FROM, from.bytes());
		buf.put_head_field(TO, to.bytes());
		buf.put_unbroken_head(&sip_data.call_id);
		buf.put_head_field(MAX_FORWARDS, b"70");
		self.cseq += 1;
		buf.put_cseq_head(self.cseq, SipMethod::Bye.as_bytes());
		buf.put_head_field(CONTENT_LENGTH, LENGTH_ZERO);
		buf.put_heads_end();

		self.last_send_bytes = Some(buf.freeze())
	}

	///生成一个100 try消息
	pub(crate) fn generate_invite_try(&self) -> Bytes {
		let sip_data = &self.sip_data;

		let mut buf = BytesMut::with_capacity(1024);
		buf.put_sip_response(&SipResponse::Try);
		buf.put_unbroken_head(&sip_data.via);
		buf.put_unbroken_head(&sip_data.from);
		buf.put_unbroken_head(&sip_data.to);
		buf.put_unbroken_head(&sip_data.call_id);
		buf.put_unbroken_head(&sip_data.cseq);
		buf.put_head_field(CONTENT_LENGTH, LENGTH_ZERO);
		buf.put_heads_end();

		buf.freeze()
	}

	/// 生成一个200的收到消息.此方法不是inv过程当中的200
	/// 此方法使用的数据为收到的上一条数据,而不是直接使用基础的数据
	pub(crate) fn generate_receive(&self, sip_data: &SipData) -> Bytes {
		let mut buf = BytesMut::with_capacity(1024);
		buf.put_sip_response(&SipResponse::Ok);
		buf.put_unbroken_head(&sip_data.via);
		buf.put_unbroken_head(&sip_data.from);
		buf.put_unbroken_head(&sip_data.to);
		buf.put_unbroken_head(&sip_data.call_id);
		buf.put_unbroken_head(&sip_data.cseq);
		buf.put_head_field(CONTENT_LENGTH, LENGTH_ZERO);
		buf.put_heads_end();

		buf.freeze()
	}

	///发送预接通
	/// 一个通话只发送一次,因此只在还没有发送过183信号的,发一个183
	pub(crate) fn pre_answer(&mut self) {
		if self.state < DialogState::PreAnswer {
			self.create_to_tag();
			self.send_to_remote(self.generate_pre_answer());
			self.state = DialogState::PreAnswer;
		}
	}

	/// 设置to_tag的值.
	///调用此方法时,如果已经存在to_tag,则不作任何改变
	/// 对应sip协议,只有在接收一个呼叫,并且发送200.300 400这类的消息时,才开始生成to_tag
	#[inline]
	pub(crate) fn create_to_tag(&mut self) {
		if self.sip_to_tag.is_empty() {
			trace!("生成一个新的to_tag");
			let mut bm = BytesMut::with_capacity(30);
			bm.extend_from_slice(b";tag=");
			bm.extend_from_slice(&generate_tag_value());
			self.sip_to_tag = bm.freeze();
		}
	}

	///生成一个180的返回信息
	pub(crate) fn generate_ringing(&self) -> Bytes {
		let sip_data = &self.sip_data;
		let mut buf = BytesMut::with_capacity(1024);

		buf.put_sip_response(&SipResponse::Ringing);
		buf.put_unbroken_head(&sip_data.via);
		buf.put_unbroken_head(&sip_data.record_route);
		buf.put_unbroken_head(&sip_data.from);
		buf.put_head_by_to_value(&self.sip_to_name, &self.sip_to_tag);
		buf.put_unbroken_head(&sip_data.call_id);
		buf.put_unbroken_head(&sip_data.cseq);
		buf.put_head_field(
			CONTACT,
			&generate_full_sip_url(
				"",
				get_bytes_str(&self.dest_num),
				get_bytes_str(&self.local_addr),
				false,
			),
		);
		buf.put_head_field(CONTENT_LENGTH, LENGTH_ZERO);
		buf.put_heads_end();

		buf.freeze()
	}

	#[inline]
	pub(crate) fn caller_num(&self) -> String {
		get_bytes_str(&self.caller_num).to_owned()
	}

	#[inline]
	pub(crate) fn dest_num(&self) -> String {
		get_bytes_str(&self.dest_num).to_owned()
	}

	pub(crate) fn generate_pre_answer(&self) -> Bytes {
		let sip_data = &self.sip_data;
		let mut buf = BytesMut::with_capacity(2048);
		self.sdp.write_to(&mut buf);
		let sdp_buf = buf.split().freeze();

		buf.put_sip_response(&SipResponse::SessionProgress);
		buf.put_unbroken_head(&sip_data.via);
		buf.put_unbroken_head(&sip_data.record_route);
		buf.put_unbroken_head(&sip_data.from);
		buf.put_head_by_to_value(&self.sip_to_name, &self.sip_to_tag);
		buf.put_unbroken_head(&sip_data.call_id);
		buf.put_unbroken_head(&sip_data.cseq);
		buf.put_head_field(
			CONTACT,
			&generate_full_sip_url(
				"",
				get_bytes_str(&self.dest_num),
				get_bytes_str(&self.local_addr),
				false,
			),
		);
		buf.put_head_field(ALLOW, sip_config().allow.as_bytes());
		buf.put_head_field(ALLOW_EVENTS, sip_config().allow_events.as_bytes());
		buf.put_head_field(SUPPORTED, sip_config().supported.as_bytes());
		buf.put_head_field(P_EARLY_MEDIA, b"sendrecv");

		buf.put_head_field(CONTENT_TYPE, b"application/sdp");
		buf.put_head_field(CONTENT_LENGTH, &sdp_buf.len().to_string().as_bytes());
		buf.put_heads_end();

		buf.extend_from_slice(&sdp_buf);

		buf.freeze()
	}

	///生成一个内部错误出去..
	///
	/// 目前直接返回请求相关数据.
	pub(crate) fn generate_internal_error(&self) -> Bytes {
		let sip_data = &self.sip_data;
		let mut buf = BytesMut::with_capacity(1024);
		buf.put_sip_response(&SipResponse::ServerInternalError);
		buf.put_unbroken_head(&sip_data.via);
		buf.put_unbroken_head(&sip_data.from);
		buf.put_head_by_to_value(&self.sip_to_name, &self.sip_to_tag);
		buf.put_unbroken_head(&sip_data.call_id);
		buf.put_unbroken_head(&sip_data.cseq);
		buf.put_head_field(CONTENT_LENGTH, LENGTH_ZERO);
		buf.put_heads_end();

		buf.freeze()
	}

	///生成一个不支持的媒体类型消息
	///返回的Bytes即是需要发送的数据.
	pub(crate) fn generate_unsupported_media_type(&self) -> Bytes {
		let sip_data = &self.sip_data;
		let mut buf = BytesMut::with_capacity(1024);
		buf.put_sip_response(&SipResponse::UnsupportedMediaType);
		buf.put_unbroken_head(&sip_data.via);
		buf.put_unbroken_head(&sip_data.from);
		buf.put_head_by_to_value(&self.sip_to_name, &self.sip_to_tag);
		buf.put_unbroken_head(&sip_data.call_id);
		buf.put_unbroken_head(&sip_data.cseq);
		buf.put_head_field(CONTENT_LENGTH, LENGTH_ZERO);
		buf.put_heads_end();

		buf.freeze()
	}
	///根据自身情况,生成一个挂断原因
	/// 这个挂断原因一般是用在inv_server的时候,需要返回一个挂断原因.
	pub(crate) fn generate_hungup_with_reason(&self, reason: EndReason) -> Bytes {
		let sip_data = &self.sip_data;

		let mut buf = BytesMut::with_capacity(1024);
		buf.put_sip_response(&reason.into());
		buf.put_unbroken_head(&sip_data.via);
		buf.put_unbroken_head(&sip_data.from);
		buf.put_head_by_to_value(&self.sip_to_name, &self.sip_to_tag);
		buf.put_unbroken_head(&sip_data.call_id);
		buf.put_unbroken_head(&sip_data.cseq);
		buf.put_head_field(CONTENT_LENGTH, LENGTH_ZERO);

		buf.put_heads_end();

		buf.freeze()
	}

	pub(crate) fn generate_answer(&self) -> Bytes {
		let sip_data = &self.sip_data;
		let mut buf = BytesMut::with_capacity(2048);
		self.sdp.write_to(&mut buf);
		let sdp_buf = buf.split().freeze();

		buf.put_sip_response(&SipResponse::Ok);
		buf.put_unbroken_head(&sip_data.via);
		buf.put_unbroken_head(&sip_data.call_id);
		buf.put_unbroken_head(&sip_data.record_route);
		buf.put_unbroken_head(&sip_data.from);
		buf.put_head_by_to_value(&self.sip_to_name, &self.sip_to_tag);
		buf.put_unbroken_head(&sip_data.cseq);
		buf.put_head_field(
			CONTACT,
			&generate_full_sip_url(
				"",
				get_bytes_str(&self.dest_num),
				get_bytes_str(&self.local_addr),
				false,
			),
		);
		buf.put_head_field(ALLOW, sip_config().allow.as_bytes());
		buf.put_head_field(SUPPORTED, sip_config().supported.as_bytes());

		buf.put_head_field(CONTENT_TYPE, b"application/sdp");
		buf.put_head_field(CONTENT_LENGTH, &sdp_buf.len().to_string().as_bytes());
		buf.put_heads_end();

		buf.extend_from_slice(&sdp_buf);

		buf.freeze()
	}

	///根据得到的sdp,找对应的匹配媒体格式..并将相应的rtp地址设置到rtp中
	/// 如果给的是一个空的数据,代表对方没有送相应的sdp包,,直接用当前自己的包即是
	pub(crate) fn set_sdp(&mut self, sdp_body: &Bytes) -> ResultNull {
		if sdp_body.is_empty() {
			return Ok(());
		}

		let peer_sdp = Sdp::from_bytes(&mut sdp_body.clone())?;
		let mut local_sdp = peer_sdp.merge(&peer_sdp)?;
		if local_sdp.m.is_empty() {
			return Err(Error::NotFound("没有匹配的media数据"));
		}
		let Some(m) = local_sdp.m.pop() else {
			return Err(Error::NotFound("未收到media的数据"));
		};

		//如果对端是一个内网,,使用sip的外网地址替换掉
		self.send_to_rtp(RtpEvent::SetPeerRtpAddr(
			self.id,
			combination_address_replace(
				peer_sdp.c.address,
				&self.peer_addr(),
				peer_sdp.first_rtp_port(),
			),
		));
		self.send_to_rtp(RtpEvent::SetPeerRtcpAddr(
			self.id,
			combination_address_replace(
				peer_sdp.c.address,
				&self.peer_addr(),
				peer_sdp.first_rtcp_port(),
			),
		));
		self.send_to_rtp(RtpEvent::SetLagAudioMedia(m.attribs));

		self.sdp = peer_sdp;
		Ok(())
	}

	#[inline]
	fn peer_addr(&self) -> IpAddr {
		self.transport.peer_addr().ip()
	}
	#[inline]
	pub(crate) fn send_to_business(&self, msg: ChannelEvent) {
		match self.to_business.try_send(msg) {
			Ok(..) => {}
			Err(TrySendError::Closed(m)) => warn!("to_business消息发送失败,已关闭 {:?}", m),
			Err(TrySendError::Full(m)) => warn!("to_business消息发送失败,传送满 {:?}", m),
		}
	}
	#[inline]
	pub(crate) fn send_to_manager(&self, msg: DialogEvent) {
		match self.to_manager.try_send(msg) {
			Ok(..) => {}
			Err(TrySendError::Closed(m)) => warn!("to_manager消息发送失败,已关闭 {:?}", m),
			Err(TrySendError::Full(m)) => warn!("to_manager消息发送失败,传送满 {:?}", m),
		}
	}
	#[inline]
	pub(crate) fn send_to_rtp(&self, msg: RtpEvent) {
		match self.to_rtp.try_send(msg) {
			Ok(..) => {}
			Err(TrySendError::Closed(m)) => error!("to_rtp消息发送失败,已关闭 {:?}", m),
			Err(TrySendError::Full(m)) => error!("to_rtp消息发送失败,传送满 {:?}", m),
		}
	}

	#[inline]
	pub(crate) fn set_tran_state(&mut self, state: SipTransactionState) {
		trace!(
			"事务状态修改.从{:?}修改为{:?}..当前事务:{:?}",
			self.tran_state,
			state,
			self.tran_handler
		);
		self.tran_state = state;
	}

	#[inline]
	pub(crate) fn tran_state(&mut self) -> SipTransactionState {
		self.tran_state
	}
}

///返回一个唯一的id值.重启或 到达u32.max() 之后会重置..不会返回0
pub(crate) fn dialog_id() -> u32 {
	static CURR_ID: AtomicU32 = AtomicU32::new(1);

	let r = CURR_ID.fetch_add(1, Ordering::Relaxed);
	match r {
		0 => CURR_ID.fetch_add(1, Ordering::Relaxed),
		_ => r,
	}
}

// ///返回dialog不存在消息
// ///
// /// 因为此消息生成时不存在对应的dialog,所以直接使用sip_data
// pub(crate) fn generate_call_not_exist(sip_data: &SipData) -> Bytes {
// 	let mut buf = BytesMut::with_capacity(1024);
// 	buf.put_sip_response(&SipResponse::DoesNotExist);
// 	buf.put_unbroken_head(&sip_data.via);
// 	buf.put_unbroken_head(&sip_data.from);
// 	buf.put_unbroken_head(&sip_data.to);
// 	buf.put_unbroken_head(&sip_data.cseq);
// 	buf.put_head_field(CONTENT_LENGTH, LENGTH_ZERO);

// 	buf.put_heads_end();

// 	buf.freeze()
// }